package com.chuke.springbootinit.job.cycle;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;


import cn.hutool.core.bean.BeanUtil;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.chuke.springbootinit.esdao.PostEsDao;
import com.chuke.springbootinit.esdao.UserEsDao;
import com.chuke.springbootinit.model.dto.post.PostEsDTO;
import com.chuke.springbootinit.model.dto.user.UserEsDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
@Slf4j
public class CanalClient {

    @Resource
    private PostEsDao postEsDao;

    @Resource
    private UserEsDao userEsDao;

    // 创建链接
    CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
            11111), "example", "", "");
    int emptyCount = 0;

    @Scheduled(fixedRate = 1000)
    private void listening() {
        int batchSize = 1000;

        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
            long batchId = message.getId();
            int size = message.getEntries().size();
            if (batchId == -1 || size == 0) {
                emptyCount++;
            } else {
                emptyCount = 0;
                // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                printEntry(message.getEntries());
            }

            connector.ack(batchId); // 提交确认
            // connector.rollback(batchId); // 处理失败, 回滚数据

        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }
            String tableName = entry.getHeader().getTableName();
            EventType eventType = rowChage.getEventType();

            log.info(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                    List<Column> afterColumnsList = rowData.getBeforeColumnsList();
                    HashMap<String, Object> map = new HashMap<>();
                    for (Column column : afterColumnsList) {
                        map.put(column.getName(), column.getValue());
                    }
                    if (tableName.equals("user")){
                        UserEsDTO userEsDTO = BeanUtil.copyProperties(map, UserEsDTO.class);
                        userEsDao.delete(userEsDTO);
                    }else if (tableName.equals("post")){
                        PostEsDTO postEsDTO = BeanUtil.copyProperties(map, PostEsDTO.class);
                        postEsDao.delete(postEsDTO);
                    }

                } else if (eventType == EventType.INSERT || eventType == EventType.UPDATE) {
                    printColumn(rowData.getAfterColumnsList());
                    List<Column> afterColumnsList = rowData.getAfterColumnsList();
                    HashMap<String, Object> map = new HashMap<>();
                    for (Column column : afterColumnsList) {
                        map.put(column.getName(), column.getValue());
                    }
                    log.info(rowData.toString());
//                    PostEsDTO postEsDTO = BeanUtil.copyProperties(map, PostEsDTO.class);
//                    postEsDao.save(postEsDTO);

                    if (tableName.equals("user")){
                        UserEsDTO userEsDTO = BeanUtil.copyProperties(map, UserEsDTO.class);
                        userEsDao.save(userEsDTO);
                    }else if (tableName.equals("post")){
                        PostEsDTO postEsDTO = BeanUtil.copyProperties(map, PostEsDTO.class);
                        postEsDao.save(postEsDTO);
                    }
                }
            }
        }
    }

    private void printColumn(List<Column> columns) {
        for (Column column : columns) {
            log.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }


}